package com.yangu.flink.cdc

import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
 * FlinkCDC使用案例并且自定义序列化方法
 * 1. 定义mysql数据源, 添加source
 *  1) mysql数据库需要开启binlog, 且binlog的日志格式需要为row类型
 *  2) 在my.cnf中配置需要开启binlog日志的数据库
 *  3) 配置mysql的四大链接参数host、port、username、password
 *  4) 配置需要监控的数据库和表
 *  5) 配置启动策略
 *  6) 配置自定义序列化策略
 * 2. 设置checkpoint
 * 3. 定义transform和sink(自己的业务逻辑)
 */
object Flink_CDCWithConsumerDeserialization {

  def main(args: Array[String]): Unit = {

    //1. 创建流式执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //1.1 设置并行度为1
    env.setParallelism(1)

    val tool = ParameterTool.fromPropertiesFile(getClass.getResource("/jdbc.properties").getPath)

    //设置mysql数据源
    val mysqlSource = MySqlSource.builder[String]
      .hostname(tool.get("jdbc.host"))
      .port(tool.getInt("jdbc.port"))
      .username(tool.get("jdbc.username"))
      .password(tool.get("jdbc.password"))
      .databaseList("gmall_flink")
      //这里如果要监控库下的所有表, 直接写*即可, 这里可以编写正则表达式来识别表名
      .tableList("gmall_flink.*")
      //关于启动策略可以根据自己的需要进行选择
      .startupOptions(StartupOptions.initial)
      //自定义反序列化器
      .deserializer(new ConsumerDeserialization)
      .build()

    val mysqlDataStream = env
      //添加mysql数据源
      .fromSource(mysqlSource, WatermarkStrategy.noWatermarks[String], "MySQL Source")
    mysqlDataStream
      //直接打印输出
      .print()

    //sink数据到Kafka
    mysqlDataStream.addSink(KafkaUtil.getKafkaSink("hadoop100:9092,hadoop101:9092,hadoop102:9092", "ods_base_db"))

    env.execute("Flink_CDCWithConsumerDeserialization Job")
  }

}
